package org.jiucheng.magpiebridge.server.aio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.StandardSocketOptions;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.CompletionHandler;
import java.text.MessageFormat;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.jiucheng.magpiebridge.protocol.Message;
import org.jiucheng.magpiebridge.server.aio.proxy.ProxyClientWriteAttachment;
import org.jiucheng.magpiebridge.server.aio.proxy.ProxyClientWriteCompletionHandler;
import org.jiucheng.magpiebridge.server.aio.proxy.ProxyServerAttachment;
import org.jiucheng.magpiebridge.server.aio.proxy.ProxyServerCompletionHandler;
import org.jiucheng.magpiebridge.util.Cfg;

/**
 * Client数据解析处理
 * 
 * @author jiucheng
 *
 */
public class ClientCompletionHandler implements CompletionHandler<Integer, ClientAttachment> {
    
    private static final Logger LOGGER = Logger.getLogger(ClientCompletionHandler.class.getName());
    // Client连接
    private static final ConcurrentMap<String, ClientAttachment> CLIENTS = new ConcurrentHashMap<String, ClientAttachment>();
    
    public void completed(Integer result, ClientAttachment attachment) {
        // client关闭连接
        if (result == -1) {
            close(attachment);
            return;
        }
        
        ByteBuffer readByteBuffer = attachment.getReadBuffer(); 
        if (readByteBuffer.position() != readByteBuffer.capacity()) {
            attachment.getAsynchronousSocketChannel().read(readByteBuffer, Server.READ_TIMEOUT, TimeUnit.SECONDS, attachment, this);
            return;
        }
        
        if (readByteBuffer.capacity() == 13) {
            // 验证消息头
            readByteBuffer.flip();
            int magic = readByteBuffer.getInt();
            byte type = readByteBuffer.get();
            int uri = readByteBuffer.getInt();
            int size = readByteBuffer.getInt();
            if (magic != Message.MAGIC) {
                close(attachment);
                return;
            }
            if (type == Message.Type.TRANSFER || type == Message.Type.AUTH) {
                readByteBuffer = ByteBuffer.allocate(size + 13);
                readByteBuffer.putInt(magic);
                readByteBuffer.put(type);
                readByteBuffer.putInt(uri);
                readByteBuffer.putInt(size);
                attachment.getAsynchronousSocketChannel().read(readByteBuffer, Server.READ_TIMEOUT, TimeUnit.SECONDS, attachment.setReadBuffer(readByteBuffer), this);
                return;
            }
        }
        
        Message message = Message.fromByteBuffer(readByteBuffer);
        byte type = message.getType();
        if (type == Message.Type.AUTH) {
            // 认证
            String clientKey = new String(message.getData());
            // 认证
            if (clientKey == null || clientKey.trim().length() < 1) {
                close(attachment);
                return;
            }
            
            try {
                initProxy(clientKey, attachment);
            } catch (IOException e) {
                e.printStackTrace();
            }
            
            readByteBuffer = ByteBuffer.allocate(13);
            attachment.getAsynchronousSocketChannel().read(readByteBuffer, Server.READ_TIMEOUT, TimeUnit.SECONDS, attachment.setReadBuffer(readByteBuffer), this);
        } else if (type == Message.Type.HEARTBEAT) {
            readByteBuffer.clear();
            attachment.getAsynchronousSocketChannel().read(readByteBuffer, Server.READ_TIMEOUT, TimeUnit.SECONDS, attachment.setReadBuffer(readByteBuffer), this);
        } else if (type == Message.Type.TRANSFER) {
            ProxyClientWriteAttachment proxyWriteAttachment = attachment.getProxy(message.getUri());
            ByteBuffer writeByteBuffer = ByteBuffer.allocate(message.getSize());
            writeByteBuffer.put(message.getData());
            writeByteBuffer.flip();
            while (!proxyWriteAttachment.writed.compareAndSet(false, true)) {}
            proxyWriteAttachment.getAsynchronousSocketChannel().write(writeByteBuffer, proxyWriteAttachment.setWriteByteBuffer(writeByteBuffer), new ProxyClientWriteCompletionHandler());
            
            readByteBuffer = ByteBuffer.allocate(13);
            attachment.getAsynchronousSocketChannel().read(readByteBuffer, Server.READ_TIMEOUT, TimeUnit.SECONDS, attachment.setReadBuffer(readByteBuffer), this);
        } else if (type == Message.Type.DISCONNECT) {
            ProxyClientWriteAttachment proxyWriteAttachment = attachment.removeProxy(message.getUri());
            if (proxyWriteAttachment != null && proxyWriteAttachment.getAsynchronousSocketChannel() != null) {
                try {
                    proxyWriteAttachment.getAsynchronousSocketChannel().close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            readByteBuffer.clear();
            attachment.getAsynchronousSocketChannel().read(readByteBuffer, Server.READ_TIMEOUT, TimeUnit.SECONDS, attachment.setReadBuffer(readByteBuffer), this);
        } else {
            readByteBuffer = ByteBuffer.allocate(13);
            attachment.getAsynchronousSocketChannel().read(readByteBuffer, Server.READ_TIMEOUT, TimeUnit.SECONDS, attachment.setReadBuffer(readByteBuffer), this);
        }
    }
    
    private synchronized static void initProxy(String clientKey, ClientAttachment attachment) throws IOException {
        ClientAttachment clientAttachment = CLIENTS.remove(clientKey);
        if (clientAttachment != null) {
            clientAttachment.close();
        }
        
        CLIENTS.put(clientKey, attachment);
        String serverMappings = Cfg.getServerMappings(clientKey);
        if (serverMappings != null && serverMappings.trim().length() > 0) {
            String[] mappings = serverMappings.split(",");
            for (String mapping : mappings) {
                String[] port2remote = mapping.split("/");
                String local = port2remote[0];
                String remote = port2remote[1];
                try {
                    AsynchronousChannelGroup group = AsynchronousChannelGroup.withThreadPool(Executors.newCachedThreadPool());
                    AsynchronousServerSocketChannel proxy = AsynchronousServerSocketChannel.open(group);
                    proxy.setOption(StandardSocketOptions.SO_REUSEADDR, true);
                    proxy.setOption(StandardSocketOptions.SO_RCVBUF, 8 * 1024 * 1024);
                    
                    String[] locals = local.split(":");
                    
                    if (LOGGER.isLoggable(Level.INFO)) {
                        LOGGER.log(Level.INFO, MessageFormat.format("----begin port={0} -> remote={1}", local, remote));
                    }
                    proxy.bind(new InetSocketAddress(locals[0], Integer.valueOf(locals[1])));
                    proxy.accept(new ProxyServerAttachment(proxy).setClientAttachment(attachment).setRemote(remote), new ProxyServerCompletionHandler());
                    if (LOGGER.isLoggable(Level.INFO)) {
                        LOGGER.log(Level.INFO, MessageFormat.format("----end port={0} -> remote={1}", local, remote));
                    }
                    attachment.addProxyServer(proxy);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public void failed(Throwable exc, ClientAttachment attachment) {
        close(attachment);
    }
    
    private void close(ClientAttachment attachment) {
        attachment.close();
    }
}
